package com.andnnl.stream;

import com.andnnl.list.Person;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Created by chenss on 2024/4/25.
 */
public class MQUtil {
    public static List<Person> pollBatch(BlockingQueue<Person> queue, int batchSize) {
        List<Person> batchSmsList = new ArrayList<>();
        for (int i = 0; i< batchSize; i++) {
            Person message = queue.poll();
            if (message == null) {
                break;
            }
            batchSmsList.add(message);
        }
        return batchSmsList;
    }

    public static List<Person> pollBatchJava8(BlockingQueue<Person> queue, int batchSize) {
        return Stream.generate(() -> queue.poll())
                .limit(batchSize)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    public static void main(String[] args) {
        BlockingQueue<Person> queue = new ArrayBlockingQueue<>(10000); // 你的队列实例
        for (int i = 0; i < 28; i++) {
            queue.offer(new Person(""+i, "张三"+i));
        }
        int batchSize = 10; // 要读取的消息数量
        List<Person> batchSmsList = MQUtil.pollBatchJava8(queue, batchSize);
        System.out.println(batchSmsList.size());
        batchSmsList.forEach(System.out::println);
    }
}
